package com.gzyj.video.handler.rtp;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.gzyj.video.common.*;
import com.gzyj.video.publisher.PublishManager;
import com.gzyj.video.utils.StackUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;


/**
 * Created by matrixy on 2019/4/9.
 */
@Slf4j
public class Jtt1078Handler extends SimpleChannelInboundHandler<Packet> {

    private static final AttributeKey<Session> SESSION_KEY = AttributeKey.valueOf("session-key");

    private int serverPort;

    private String version = "1078";

    public Jtt1078Handler(int serverPort, String version) {

        this.serverPort = serverPort;
        if (StringUtils.isNotEmpty(version))
            this.version = version;
    }

    private void sendStaus(String tag,String msg){
        ChannelHandlerContext ctx= WebSocketStatsSession.storage.get(tag);

        if(ctx!=null){
            io.netty.channel.Channel cl= ctx.channel();

            if (!cl.isActive()) {
                System.out.println(tag + "已断开状态连接");
            }else{
                cl.writeAndFlush(new TextWebSocketFrame("心跳"));
            }

        }else{
            System.out.println("不存在状态连接"+tag);
        }

    }


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Packet packet) {
        try {
            io.netty.channel.Channel nettyChannel = ctx.channel();

            packet.seek(8);
            if (this.version.equalsIgnoreCase("1078")) {
                // 苏标或1078
                String sim = packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD();
                int channel = packet.nextByte() & 0xff;
                String tag = sim + "-" + channel;

                if (SessionManager.contains(nettyChannel, "tag") == false) {
                    Channel chl = PublishManager.getInstance().open(tag);
                    SessionManager.set(nettyChannel, "tag", tag);
                    log.info("start publishing: {} -> {}-{}", Long.toHexString(chl.hashCode() & 0xffffffffL), sim, channel);
                }

                Integer sequence = SessionManager.get(nettyChannel, "video-sequence");
                if (sequence == null) sequence = 0;
                // 1. 做好序号
                // 2. 音频需要转码后提供订阅
                int lengthOffset = 28;
                int dataType = (packet.seek(15).nextByte() >> 4) & 0x0f;
                int pkType = packet.seek(15).nextByte() & 0x0f;
                // 透传数据类型：0100，没有后面的时间以及Last I Frame Interval和Last Frame Interval字段
                if (dataType == 0x04) lengthOffset = 28 - 8 - 2 - 2;
                    // 音频数据 没有Last I Frame Interval和Last Frame Interval字段
                else if (dataType == 0x03) lengthOffset = 28 - 2 - 2;

                int pt = packet.seek(5).nextByte() & 0x7f;// 负载类型

                if (dataType == 0x00 || dataType == 0x01 || dataType == 0x02) {
                    // 碰到结束标记时，序号+1
                    if (pkType == 0 || pkType == 2) {
                        sequence += 1;
                        SessionManager.set(nettyChannel, "video-sequence", sequence);
                    }
                    long timestamp = packet.seek(16).nextLong();
                    PublishManager.getInstance().publishVideo(tag, sequence, timestamp, pt, packet.seek(lengthOffset + 2).nextBytes());

                    sendStaus(tag,"");
                } else if (dataType == 0x03) {
                    long timestamp = packet.seek(16).nextLong();
                    byte[] data = packet.seek(lengthOffset + 2).nextBytes();
                    PublishManager.getInstance().publishAudio(tag, sequence, timestamp, pt, data);

                    sendStaus(tag,"");

                }
            } else {
                String sim = packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD();

                // 逻辑通道号
                int channel = packet.nextByte() & 0xff;
                String tag = sim + "-" + channel;
                if (SessionManager.contains(nettyChannel, "tag") == false) {
                    Channel chl = PublishManager.getInstance().open(tag);
                    SessionManager.set(nettyChannel, "tag", tag);
                    log.info("start publishing: {} -> {}-{}", Long.toHexString(chl.hashCode() & 0xffffffffL), sim, channel);
                }

                Integer sequence = SessionManager.get(nettyChannel, "video-sequence");
                if (sequence == null) sequence = 0;

                // 1. 做好序号
                // 2. 音频需要转码后提供订阅
                int lengthOffset = 32;
                //数据类型
                int dataType = (packet.seek(19).nextByte() >> 4) & 0x0f;
                //分包标记
                int pkType = packet.seek(19).nextByte() & 0x0f;

                // 透传数据类型：0100，没有后面的时间以及Last I Frame Interval和Last Frame Interval字段
                if (dataType == 0x04) lengthOffset = 32 - 8 - 2 - 2;
                    // 音频数据 没有Last I Frame Interval和Last Frame Interval字段
                else if (dataType == 0x03) lengthOffset = 32 - 2 - 2;

                int pt = packet.seek(5).nextByte() & 0x7f;// 负载类型


                if (dataType == 0x00 || dataType == 0x01 || dataType == 0x02) {
                    // 碰到结束标记时，序号+1
                    if (pkType == 0 || pkType == 2) {
                        sequence += 1;
                        SessionManager.set(nettyChannel, "video-sequence", sequence);
                    }
                    long timestamp = packet.seek(20).nextLong();
                    PublishManager.getInstance().publishVideo(tag, sequence, timestamp, pt, packet.seek(lengthOffset + 2).nextBytes());
                } else if (dataType == 0x03) {
                    //音频
                    long timestamp = packet.seek(20).nextLong();
                    byte[] data = packet.seek(lengthOffset + 2).nextBytes();
                    PublishManager.getInstance().publishAudio(tag, sequence, timestamp, pt, data);
                }


            }
        } catch (Exception e) {
            log.error("解析报错 \n\n {} ", StackUtils.getStackInfo(e));
        }


    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        String tag = SessionManager.get(ctx.channel(), "tag");
        if(tag!=null){
            Channel cl= PublishManager.getInstance().channels.get(tag);
            cl.close();
            PublishManager.getInstance().channels.remove(tag);
            log.error(tag +" "+cl.isPublishing()+ " 断开连接");
            JSONObject param=new JSONObject();
            param.put("tag",tag);
            param.put("status","stop");
            WebSocketStatsSession.storage.get(tag).writeAndFlush(new TextWebSocketFrame(param.toJSONString()));
            release(ctx.channel());
        }

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // super.exceptionCaught(ctx, cause);
        if (!ctx.channel().isActive()) {
            release(ctx.channel());
            ctx.close();
            log.error("释放连接 解析协议报错 \n {}", StackUtils.getStackInfo(cause));
        } else {
//            ctx.fireExceptionCaught(cause);
            log.error("解析协议报错 \n {}", StackUtils.getStackInfo(cause));
        }

    }

    private int times;

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        /*if (IdleStateEvent.class.isAssignableFrom(evt.getClass())) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                String tag = SessionManager.get(ctx.channel(), "tag");
                log.info("read timeout: {}", tag);
                release(ctx.channel());
            }
        }*/
        IdleStateEvent event = (IdleStateEvent) evt;
        String eventDesc = null;
        switch (event.state()) {
            case READER_IDLE:
                eventDesc = "读空闲";
                break;
            case WRITER_IDLE:
                eventDesc = "写空闲";
                break;
            case ALL_IDLE:
                eventDesc = "读写空闲";
                break;
        }
        log.error(ctx.channel().remoteAddress() + "发生超时事件--" + eventDesc);
        times++;
        if (times > 3) {
            System.out.println("空闲次数超过三次 关闭连接");
            ctx.writeAndFlush("you are out");
            release(ctx.channel());
        }


    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }


    private void release(io.netty.channel.Channel channel) {
        String tag = SessionManager.get(channel, "tag");
        if (tag != null) {
            log.info("close netty channel: {}", tag);
            PublishManager.getInstance().close(tag);
        }
    }
}
